feat(backend): checkpointed, reorg-aware ledger ingestion pipeline - #392

Summary

Closes #319.

This repo had no ingestion pipeline for Stellar ledger data yet. Adds backend/src/ingestion/, which pulls ledger data and writes derived corridor/anchor/network rows (RawSnapshotRow, already used by backend/src/snapshot) in a way that:

• survives being re-run over an already-processed range (byte-identical result), and
• survives a Horizon/RPC reorg, self-healing with no manual intervention.

Design

• watermark.rs — the checkpoint is (ledger_sequence, ledger_hash), not sequence alone, so a reorg is detectable directly: the next fetched ledger's prev_hash stops matching the checkpoint's hash. WatermarkStore is a trait with an InMemoryWatermarkStore reference implementation, matching this codebase's existing pattern (AnalyticsSink/InMemoryAnalyticsSink, the reconciliation module's stores) rather than introducing a new database dependency this crate doesn't otherwise have.
• upsert.rs — every derived write is a full replace keyed on (ledger_sequence, entity_id) where entity_id = (source, corridor). Re-ingesting a range, or rolling one back after a reorg and re-ingesting it, converges on the same stored state instead of accumulating duplicates or leaving orphaned rows behind from an abandoned fork.
• fetch.rs — LedgerSource trait with two implementations:
  • HorizonLedgerSource: a real Horizon-backed wrapper (via reqwest), reusing network::NetworkClient for its URL config. It derives one coarse, ledger-level RawSnapshotRow from each ledger's transaction success/failure counts — intentionally a proxy, not the full per-corridor asset-pair breakdown a real analytics engine would compute from payment operations (that computation doesn't exist in this codebase yet — see Core payment-reliability/latency-percentile computation engine does not exist #383). This wrapper's job is to prove the checkpointing/reorg machinery works end-to-end with a real row flowing through it.
  • FakeLedgerSource: deterministic, in-memory, for tests. fork_from() builds a chain whose hash linkage genuinely diverges from the original at the fork point (not a canned response), so the pipeline's reorg detection is actually exercised, not simulated.
• mod.rs — IngestionPipeline orchestrates fetch → reorg-check → upsert → checkpoint. On a hash mismatch it rolls the watermark back by a confirmation depth (default N=3), invalidates every derived row from there forward, and resumes — no manual intervention needed. If the reorg is deeper than the pipeline's configured start ledger, it resets ingestion entirely rather than getting stuck. run_forever logs and retries past a failed tick instead of dying permanently on the first transient error (worth calling out since ReconciliationJob::run_forever dies permanently on the first transient error #380, open in this same repo, is exactly that anti-pattern in the reconciliation job — this pipeline deliberately doesn't repeat it).
• reconcile.rs — ReconciliationCheck runs on a schedule, recomputing a checksum straight from raw ledger data (refetched from LedgerSource) and comparing it against what's actually stored, so a bug in the ingestion write path itself — not just an outage — gets caught. The checksum combines rows with wrapping addition rather than XOR, deliberately: XOR would let two occurrences of the same row cancel out, silently hiding exactly the double-counting bug this check exists to catch.

Testing

• cargo fmt / cargo clippy (including the strict --lib flags this repo's CI uses: -D clippy::unwrap_used -D clippy::expect_used -D clippy::panic -D warnings) are clean for every file this PR touches. (Note: cargo clippy --lib currently fails on unmodified main with ~51 pre-existing errors in unrelated files — verified via git stash before this diff existed — and two existing integration tests, connected_trace_test.rs and mixed_version_replay_test.rs, already fail to compile on main due to a pre-existing soroban-sdk/stellar-xdr version conflict. None of that is from this PR; out of scope to fix here.)
• 27 unit tests across the new module (checkpoint round-trip, idempotent upsert incl. dropped-entity and re-upsert-with-different-value cases, reorg rollback depth, the deeper-than-start-sequence reset path, checksum order-independence and duplicate-sensitivity).
• backend/tests/ingestion_reorg_test.rs — the acceptance test: replays a 7-ledger fixture where a reorg at ledger 5 replaces two ledgers with a 3-ledger competing fork, and asserts the final state matches only the canonical chain with zero orphaned or duplicated rows, using nothing but the pipeline's normal run_once() poll loop.
• All pre-existing lib tests (54) and integration tests that already compiled on main still pass unmodified.

Test plan

• cargo build --lib
• cargo clippy (strict flags) clean on every file this PR touches
• cargo fmt --check clean on every file this PR touches
• 27 new unit tests pass
• ingestion_reorg_test.rs (the reorg fixture + byte-identical re-run acceptance tests) pass
• Full existing cargo test --lib (54 tests) and other working integration test files still pass — no regressions